package com.zdb.demo.flink.streaming;

import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.io.TextValueInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.types.StringValue;
import org.apache.flink.util.Collector;

import java.util.stream.Stream;

public class StreamingWordCount {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration(){{
            setInteger("rest.port", 9191);
            setBoolean("local.start-webserver", true);
        }};
        final StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        streamEnv.setParallelism(4);
        String feedFileFolderPath = "/E:/logs/tmp";
        org.apache.flink.core.fs.Path filePath = new org.apache.flink.core.fs.Path(feedFileFolderPath);

        TextValueInputFormat inputFormat = new TextValueInputFormat(new Path("aaa"));

        DataStream<StringValue> inputStream = streamEnv.readFile(inputFormat, feedFileFolderPath, FileProcessingMode.PROCESS_CONTINUOUSLY, 1000);

//        streamEnv.readTextFile("/E:/logs/tmp", "utf-8")

//        inputStream.flatMap(new FlatMapFunction<StringValue, Tuple2<String, Integer>>() {
//            @Override
//            public void flatMap(StringValue s, Collector<Tuple2<String, Integer>> collector) throws Exception {
//                Stream.of(s.getValue().split(",")).forEach(word -> collector.collect(new Tuple2<>(word, 1)));
//            }
//        })
        inputStream.flatMap(new RichFlatMapFunction<StringValue, Tuple2<String, Integer>>() {

            private LongCounter lineCounter = new LongCounter(0);
            private LongCounter wordCounter = new LongCounter(0);

            @Override
            public void open(Configuration parameters) throws Exception {
                getRuntimeContext().addAccumulator("line-number", lineCounter);
                getRuntimeContext().addAccumulator("word-number", wordCounter);
            }

            @Override
            public void flatMap(StringValue s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                lineCounter.add(1);
                Stream.of(s.getValue().split(",")).forEach(word -> {
                    collector.collect(new Tuple2<>(word, 1));
                    if(word != null && word.length() > 10) {
                        wordCounter.add(1l);
                    }
                });
            }
        })
//                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0).timeWindow(Time.seconds(20), Time.seconds(5)).sum(1).writeAsCsv("/E:/logs/result.csv");

        streamEnv.execute("file word count");
    }
}
